package cn.seaboot.commons.thread;

import cn.seaboot.commons.lang.Warning;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * 工作线程
 * <p>
 * 起一个线程，不断从队列中取出数据进行加工
 *
 * @param <T> 上下文
 * @param <U> 待处理数据
 * @author Mr.css
 * @version 2025-05-07 14:13
 */
public class WorkRunnable<T, U> implements Runnable {
    private final Logger logger = LoggerFactory.getLogger(WorkRunnable.class);

    private final WorkTask<T,U > task;

    WorkRunnable(WorkTask<T, U> task) {
        this.task = task;
    }

    /**
     * 哑元元素，需要注意所有工作线程共享这个元素
     */
    public static final Object DUMB = new Object();

    /**
     * 上下文
     */
    private T context;
    /**
     * 释放资源程序
     */
    private Consumer<T> release;

    /**
     * 元素处理程序
     */
    private BiConsumer<T, U> consumer;

    /**
     * 异常处理程序
     * <p>
     * 如果继续抛出异常，则可以提前中断工作线程
     */
    private BiConsumer<Exception, U> fullback;

    /**
     * 阻塞队列
     */
    private BlockingQueue<Object> queue;

    public BlockingQueue<Object> getQueue() {
        return queue;
    }

    public void setQueue(BlockingQueue<Object> queue) {
        this.queue = queue;
    }

    public T getContext() {
        return context;
    }

    public void setContext(T context) {
        this.context = context;
    }

    public Consumer<T> getRelease() {
        return release;
    }

    public void setRelease(Consumer<T> release) {
        this.release = release;
    }

    public BiConsumer<T, U> getConsumer() {
        return consumer;
    }

    public void setConsumer(BiConsumer<T, U> consumer) {
        this.consumer = consumer;
    }

    public BiConsumer<Exception, U> getFullback() {
        return fullback;
    }

    public void setFullback(BiConsumer<Exception, U> fullback) {
        this.fullback = fullback;
    }

    /**
     * 添加元素
     * <p>
     * 会自动阻塞线程，不必关注线程容量问题
     *
     * @param t 元素
     * @throws InterruptedException -
     */
    public void put(T t) throws InterruptedException {
        this.queue.put(t);
    }

    /**
     * 放置哑元元素，轮询到哑元元素的时候，退出循环，
     * <p>
     * 如果多个 {@link WorkRunnable} 绑定同一个队列，则随机停止其中一个
     *
     * @throws InterruptedException -
     */
    public void putDumb() throws InterruptedException {
        this.queue.put(DUMB);
    }

    @Override
    @SuppressWarnings(Warning.UNCHECKED)
    public void run() {
        try {
            for (; ; ) {
                U msg = (U) queue.take();
                if (DUMB == msg) {
                    break;
                } else {
                    try {
                        // handler message
                        consumer.accept(context, msg);
                    } catch (Exception e) {
                        // on failed
                        this.fullback.accept(e, msg);
                    }
                }
            }
            logger.debug("{} work complete!", Thread.currentThread());
        } catch (Exception e) {
            logger.error("{} working exception!", Thread.currentThread(), e);
        } finally {
            try {
                // release resources
                if (this.release != null) {
                    this.release.accept(context);
                }
            } catch (Exception e) {
                logger.error("{} release failed: {}", Thread.currentThread(), context, e);
            } finally {
                // release lock
                this.task.countDown();
            }
        }
    }
}